package com.atguigu.flink.datastramapi.transform;

import com.atguigu.flink.pojo.OrderEvent;
import com.atguigu.flink.pojo.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * Created by Smexy on 2023/2/25
 */
public class Demo12_ConnectExec
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //读取数据
        SingleOutputStreamOperator<OrderEvent> orderDs = env
            .readTextFile("data/OrderLog.csv")
            .map(new MapFunction<String, OrderEvent>()
            {
                @Override
                public OrderEvent map(String value) throws Exception {
                    String[] words = value.split(",");
                    return new OrderEvent(
                        Long.valueOf(words[0]),
                        words[1],
                        words[2],
                        Long.valueOf(words[3])
                    );
                }
            });

        SingleOutputStreamOperator<TxEvent> txDS = env
            .readTextFile("data/ReceiptLog.csv")
            .map(new MapFunction<String, TxEvent>()
            {
                @Override
                public TxEvent map(String value) throws Exception {
                    String[] words = value.split(",");
                    return new TxEvent(
                        words[0],
                        words[1],
                        Long.valueOf(words[2])
                    );
                }
            });

        //对账的前提是要把两个流联合在一起
        orderDs.connect(txDS)
               .keyBy(OrderEvent::getTxId,TxEvent::getTxId)
               // K, IN1, IN2, OUT
               .process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>()
               {
                   //借助属性读取，称为缓存
                   // OrderEvent 需要缓存 txId:orderId
                   private Map<String,Long> orderEventCache=new HashMap<>();

                   // TxEvent 需要缓存 txId
                   private Set<String> txEventCache = new HashSet<>();

                   /*
                    处理orderDs中的数据。  处理的逻辑，每种数据到达后，会先去对方的缓存中读取。读到，就对账成功。
                                        读不到？
                                                把当前数据写入缓存，等待后续到达的对方数据匹配。
                                                    后续，对方数据到达了，就对账成功。
                                                    后续，对象数据缺失，对账不成功，就不会输出。
                    */
                   @Override
                   public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
                       String txId = value.getTxId();
                       Long orderId = value.getOrderId();
                       //从txEvent的缓存中读取
                       if (txEventCache.contains(txId)) {
                           out.collect(orderId +" 对账成功...");
                       }else {
                           //订单早到了，或压根没有支付信息。将自己写入缓存，等待后续可能到达的txEvent配对
                           orderEventCache.put(txId,orderId);

                       }
                   }

                   //处理txDS中的数据
                   @Override
                   public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
                       Long orderId = orderEventCache.get(value.getTxId());
                       if (orderId != null){
                           out.collect(orderId +" 对账成功...");
                       }else{
                           //把自己写入缓存，等待后续可能到达的OrderEvent配对
                           txEventCache.add(value.getTxId());
                       }

                   }
               })
               .print();



        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
